TCP 连接在内核中是如何管理的

基础概念题

1. TCP 连接在内核中是以什么数据结构管理的？请详细解释其组成

参考答案：

TCP 连接在内核中主要通过 套接字（Socket） 数据结构进行管理，核心包括：

核心组件详解：

1. Socket 结构体：

struct sock {
    struct sock_common __sk_common;  // 连接四元组信息
    socket_lock_t sk_lock;           // 锁机制
    struct sk_buff_head sk_receive_queue;  // 接收队列
    struct sk_buff_head sk_write_queue;    // 发送队列
    int sk_state;                    // TCP 状态
    // ... 更多字段
};

2. TCP Control Block (TCB)：

struct tcp_sock {
    struct inet_connection_sock inet_conn;
    u32 snd_nxt;    // 下一个要发送的序列号
    u32 snd_una;    // 未确认的最小序列号
    u32 rcv_nxt;    // 期望接收的下一个序列号
    u32 rcv_wnd;    // 接收窗口大小
    // ... 拥塞控制、重传等字段
};

面试追问：Go 语言中的 net.Conn 是如何与内核 socket 关联的？

2. 请解释 TCP 连接的四元组是什么，内核如何基于四元组管理连接？

参考答案：

四元组组成：

• 源 IP 地址 (Source IP)
• 源端口号 (Source Port)
• 目标 IP 地址 (Destination IP)
• 目标端口号 (Destination Port)

内核管理机制：

哈希表查找过程：

// 简化的内核查找逻辑
struct sock *tcp_lookup_established(struct net *net,
                                   __be32 saddr, __be16 sport,
                                   __be32 daddr, __be16 dport) {
    unsigned int hash = tcp_hashfn(saddr, sport, daddr, dport);
    struct sock *sk;
    
    // 在已建立连接的哈希表中查找
    sk_for_each(sk, &tcp_hashinfo.ehash[hash]) {
        if (sk->sk_daddr == saddr && sk->sk_sport == sport &&
            sk->sk_rcv_saddr == daddr && sk->sk_dport == dport) {
            return sk;
        }
    }
    return NULL;
}

Go 语言中的应用：

func main() {
    // 客户端四元组：本机IP:随机端口 -> 服务器IP:8080
    conn, err := net.Dial("tcp", "192.168.1.100:8080")
    if err != nil {
        log.Fatal(err)
    }
    
    // 打印本地和远程地址
    fmt.Printf("Local: %s, Remote: %s\n", 
               conn.LocalAddr(), conn.RemoteAddr())
}

3. TCP 连接的状态机有哪些状态？请画出状态转换图并解释关键转换

TCP 状态机：

关键状态解释：

状态	说明	对应场景
LISTEN	服务器等待连接	net.Listen()
SYN_SENT	客户端发送连接请求	net.Dial() 过程中
ESTABLISHED	连接已建立	正常数据传输
TIME_WAIT	等待可能延迟的分组	主动关闭方等待 2MSL
CLOSE_WAIT	等待应用层关闭	被动关闭方

面试重点：TIME_WAIT 状态的作用和问题

4. 什么是 TIME_WAIT 状态？为什么需要 2MSL 等待时间？在高并发场景下可能带来什么问题？

TIME_WAIT 存在的原因：

2MSL 等待的作用：

1. 确保最后的 ACK 到达：

   • 如果最后的 ACK 丢失，服务器会重传 FIN
   • 客户端需要能够重新发送 ACK

2. 防止旧连接的数据包干扰新连接：

   • 确保网络中该四元组的所有数据包都消失
   • MSL（Maximum Segment Lifetime）通常为 30 秒到 2 分钟

高并发场景的问题：

// 问题场景：短连接高并发
func problematicClient() {
    for i := 0; i < 10000; i++ {
        conn, err := net.Dial("tcp", "server:8080")
        if err != nil {
            log.Printf("连接失败: %v", err)
            continue
        }
        
        // 快速处理后关闭
        conn.Write([]byte("quick request"))
        conn.Close() // 主动关闭，进入 TIME_WAIT
    }
    // 可能导致端口耗尽！
}

问题表现：

• 端口耗尽：大量端口处于 TIME_WAIT 状态
• 内存消耗：每个 TIME_WAIT 连接占用内存
• 连接建立失败：bind: address already in use

解决方案：

# 调整内核参数
echo 1 > /proc/sys/net/ipv4/tcp_tw_reuse    # 重用 TIME_WAIT 端口
echo 1 > /proc/sys/net/ipv4/tcp_tw_recycle  # 快速回收（已废弃）

# 调整 TIME_WAIT 超时时间
echo 30 > /proc/sys/net/ipv4/tcp_fin_timeout

内核数据结构深度解析

5. 请详细解释 TCP 的发送缓冲区和接收缓冲区是如何工作的？

缓冲区结构：

发送缓冲区详解：

Go 语言中的缓冲区操作：

func demonstrateBuffering() {
    conn, _ := net.Dial("tcp", "localhost:8080")
    defer conn.Close()
    
    // 写入数据到发送缓冲区
    data := make([]byte, 1024*1024) // 1MB 数据
    n, err := conn.Write(data)
    if err != nil {
        log.Printf("写入错误: %v", err)
    }
    
    // Write 返回不意味着数据已发送到网络
    // 只是写入了内核发送缓冲区
    fmt.Printf("写入了 %d 字节到缓冲区\n", n)
    
    // 设置缓冲区大小
    if tcpConn, ok := conn.(*net.TCPConn); ok {
        tcpConn.SetWriteBuffer(64 * 1024)  // 64KB 发送缓冲区
        tcpConn.SetReadBuffer(64 * 1024)   // 64KB 接收缓冲区
    }
}

缓冲区管理机制：

操作	发送缓冲区	接收缓冲区
写入满了	write() 阻塞或返回 EAGAIN	丢弃数据包或发送零窗口
读取空了	正常（等待应用写入）	read() 阻塞或返回 EAGAIN
自动调整	根据拥塞控制调整发送窗口	根据处理能力调整接收窗口

6. 什么是 TCP 的滑动窗口机制？内核如何实现流量控制？

滑动窗口原理：

详细的滑动窗口工作流程：

内核实现机制：

// TCP 发送窗口管理（简化版）
static unsigned int tcp_snd_wnd_test(struct tcp_sock *tp, 
                                     struct sk_buff *skb) {
    u32 end_seq = TCP_SKB_CB(skb)->end_seq;
    
    // 检查是否超出发送窗口
    if (after(end_seq, tp->snd_una + tp->snd_wnd))
        return 0;  // 超出窗口，不能发送
        
    return 1;  // 可以发送
}

// 接收窗口更新
static void tcp_update_rcv_wnd(struct tcp_sock *tp) {
    int free_space = tcp_space(sk);  // 计算可用缓冲区空间
    
    if (free_space < tp->rcv_wnd / 4) {
        tp->rcv_wnd = free_space;  // 缩小窗口
    }
}

Go 语言中观察窗口行为：

func observeWindowBehavior() {
    // 创建一个小缓冲区的连接
    conn, _ := net.Dial("tcp", "localhost:8080")
    defer conn.Close()
    
    if tcpConn, ok := conn.(*net.TCPConn); ok {
        // 设置小的接收缓冲区来观察窗口效应
        tcpConn.SetReadBuffer(1024)
    }
    
    // 发送大量数据观察阻塞行为
    data := make([]byte, 1024*1024)
    start := time.Now()
    
    n, err := conn.Write(data)
    duration := time.Since(start)
    
    fmt.Printf("写入 %d 字节耗时 %v\n", n, duration)
    // 如果接收方处理慢，这里会因为窗口限制而阻塞
}

7. TCP 的拥塞控制是如何实现的？请解释慢启动、拥塞避免、快重传和快恢复

拥塞控制状态机：

各阶段详细说明：

1. 慢启动阶段：

2. 拥塞避免阶段：

内核实现（简化版）：

// TCP 拥塞控制实现
void tcp_cong_avoid(struct sock *sk, u32 ack, u32 acked) {
    struct tcp_sock *tp = tcp_sk(sk);
    
    if (!tcp_is_cwnd_limited(sk))
        return;
    
    if (tp->snd_cwnd <= tp->snd_ssthresh) {
        // 慢启动：指数增长
        tp->snd_cwnd += acked;
    } else {
        // 拥塞避免：线性增长
        tp->snd_cwnd += max(1U, acked * tp->mss_cache / tp->snd_cwnd);
    }
}

// 快重传检测
void tcp_fast_retrans_alert(struct sock *sk, int pkts_acked,
                           int prior_packets) {
    struct tcp_sock *tp = tcp_sk(sk);
    
    if (tp->dup_acks >= 3) {  // 收到3个重复ACK
        tcp_enter_recovery(sk);  // 进入快恢复
        tcp_fast_retrans(sk);    // 快重传
    }
}

Go 语言中观察拥塞控制：

func observeCongestionControl() {
    // 通过调整网络延迟和丢包来观察拥塞控制
    conn, _ := net.Dial("tcp", "remotehost:8080")
    defer conn.Close()
    
    // 发送大文件来触发拥塞控制
    file, _ := os.Open("largefile.dat")
    defer file.Close()
    
    start := time.Now()
    written, err := io.Copy(conn, file)
    duration := time.Since(start)
    
    throughput := float64(written) / duration.Seconds() / 1024 / 1024
    fmt.Printf("传输 %d 字节，吞吐量: %.2f MB/s\n", written, throughput)
}

连接建立与关闭详解

8. 请详细解释 TCP 三次握手过程，以及为什么需要三次而不是两次？

三次握手详细流程：

为什么需要三次握手？

1. 防止旧连接请求：

2. 确认双方接收能力：
   • 第一次：服务器确认客户端发送能力
   • 第二次：客户端确认服务器接收和发送能力
   • 第三次：服务器确认客户端接收能力

Go 语言中的连接建立：

func explainConnect() {
    // net.Dial 内部执行三次握手
    start := time.Now()
    conn, err := net.Dial("tcp", "google.com:80")
    duration := time.Since(start)
    
    if err != nil {
        log.Printf("连接失败: %v", err)
        return
    }
    defer conn.Close()
    
    fmt.Printf("连接建立耗时: %v\n", duration)
    // 这个时间包括了 DNS 解析 + 三次握手的时间
    
    // 可以通过 DialTimeout 设置连接超时
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    
    var dialer net.Dialer
    conn2, err := dialer.DialContext(ctx, "tcp", "slow-server.com:80")
    if err != nil {
        log.Printf("连接超时: %v", err)
    }
    defer conn2.Close()
}

9. TCP 四次挥手的过程是怎样的？为什么需要四次而不是三次？

四次挥手详细流程：

为什么需要四次挥手？

TCP 是全双工通信，需要分别关闭两个方向的数据流：

1. 第一、二次挥手：关闭客户端到服务器的数据流
2. 第三、四次挥手：关闭服务器到客户端的数据流

关键理解：

• 第二次和第三次挥手之间，服务器可能还有数据要发送
• 这个阶段称为"半关闭"状态

Go 语言中的连接关闭：

func demonstrateClose() {
    conn, _ := net.Dial("tcp", "localhost:8080")
    
    // 正常关闭：调用 Close() 会发起四次挥手
    defer conn.Close()
    
    // 可以单独关闭读或写方向
    if tcpConn, ok := conn.(*net.TCPConn); ok {
        // 关闭写方向（发送FIN）
        tcpConn.CloseWrite()
        
        // 此时仍可以读取服务器发送的数据
        buffer := make([]byte, 1024)
        n, err := conn.Read(buffer)
        if err != nil {
            fmt.Printf("读取到 %d 字节: %s\n", n, string(buffer[:n]))
        }
        
        // 关闭读方向
        tcpConn.CloseRead()
    }
}

// 检测连接关闭
func detectConnectionClose() {
    conn, _ := net.Dial("tcp", "localhost:8080")
    defer conn.Close()
    
    go func() {
        buffer := make([]byte, 1024)
        for {
            n, err := conn.Read(buffer)
            if err != nil {
                if err == io.EOF {
                    fmt.Println("对方关闭了连接")
                } else {
                    fmt.Printf("读取错误: %v\n", err)
                }
                return
            }
            fmt.Printf("接收到 %d 字节\n", n)
        }
    }()
    
    time.Sleep(10 * time.Second)
}

10. 什么情况下会出现大量 CLOSE_WAIT 和 TIME_WAIT 状态？如何解决？

CLOSE_WAIT 状态分析：

CLOSE_WAIT 堆积的原因：

// 有问题的服务器代码
func problematicServer() {
    listener, _ := net.Listen("tcp", ":8080")
    
    for {
        conn, _ := listener.Accept()
        go func(c net.Conn) {
            // 没有 defer c.Close()!
            buffer := make([]byte, 1024)
            for {
                n, err := c.Read(buffer)
                if err != nil {
                    // 出错时没有关闭连接就直接返回
                    // 导致连接泄露，处于 CLOSE_WAIT 状态
                    fmt.Printf("读取错误: %v\n", err)
                    return  // Bug: 没有关闭连接
                }
                c.Write(buffer[:n])
            }
        }(conn)
    }
}

// 正确的处理方式
func correctServer() {
    listener, _ := net.Listen("tcp", ":8080")
    
    for {
        conn, _ := listener.Accept()
        go func(c net.Conn) {
            defer c.Close()  // 确保连接被关闭
            
            buffer := make([]byte, 1024)
            for {
                n, err := c.Read(buffer)
                if err != nil {
                    if err != io.EOF {
                        fmt.Printf("读取错误: %v\n", err)
                    }
                    return  // defer 会关闭连接
                }
                c.Write(buffer[:n])
            }
        }(conn)
    }
}

TIME_WAIT 状态问题：

高并发短连接场景中，大量连接处于 TIME_WAIT 状态：

// 产生大量 TIME_WAIT 的场景
func createManyTimeWait() {
    for i := 0; i < 10000; i++ {
        conn, err := net.Dial("tcp", "localhost:8080")
        if err != nil {
            fmt.Printf("连接 %d 失败: %v\n", i, err)
            continue
        }
        
        // 快速处理
        conn.Write([]byte("quick data"))
        
        // 主动关闭连接，进入 TIME_WAIT
        conn.Close()
        
        if i%1000 == 0 {
            fmt.Printf("完成 %d 个连接\n", i)
        }
    }
}

解决方案：

1. 应用层解决方案：

// 连接池复用
type ConnectionPool struct {
    pool chan net.Conn
    addr string
}

func (p *ConnectionPool) Get() (net.Conn, error) {
    select {
    case conn := <-p.pool:
        return conn, nil
    default:
        return net.Dial("tcp", p.addr)
    }
}

func (p *ConnectionPool) Put(conn net.Conn) {
    select {
    case p.pool <- conn:
    default:
        conn.Close()
    }
}

// 长连接 + HTTP/1.1 Keep-Alive
func useLongConnection() {
    client := &http.Client{
        Transport: &http.Transport{
            MaxIdleConns:        100,
            MaxIdleConnsPerHost: 10,
            IdleConnTimeout:     90 * time.Second,
        },
    }
    
    // 复用连接进行多次请求
    for i := 0; i < 1000; i++ {
        resp, err := client.Get("http://localhost:8080/api")
        if err == nil {
            resp.Body.Close()
        }
    }
}

2. 系统参数调优：

# 允许重用 TIME_WAIT 状态的端口
echo 1 > /proc/sys/net/ipv4/tcp_tw_reuse

# 减少 TIME_WAIT 超时时间
echo 30 > /proc/sys/net/ipv4/tcp_fin_timeout

# 增加本地端口范围
echo "1024 65535" > /proc/sys/net/ipv4/ip_local_port_range

# 监控当前连接状态
ss -ant | awk '/TIME_WAIT/ {count++} END {print "TIME_WAIT:", count+0}'
ss -ant | awk '/CLOSE_WAIT/ {count++} END {print "CLOSE_WAIT:", count+0}'

性能优化与监控

11. 如何监控和调试 TCP 连接的状态？有哪些重要的性能指标？

TCP 连接监控工具：

1. 系统级监控：

# 查看连接状态统计
ss -s

# 查看具体连接信息
ss -tuln | head -20

# 实时监控连接变化
watch -n 1 'ss -s'

# 查看 TCP 内存使用
cat /proc/net/sockstat

2. Go 程序内监控：

package main

import (
    "fmt"
    "net"
    "runtime"
    "time"
)

// TCP 连接监控结构
type ConnectionMonitor struct {
    activeConnections int64
    totalConnections  int64
    startTime        time.Time
}

func (cm *ConnectionMonitor) handleConnection(conn net.Conn) {
    defer func() {
        conn.Close()
        atomic.AddInt64(&cm.activeConnections, -1)
    }()
    
    atomic.AddInt64(&cm.activeConnections, 1)
    atomic.AddInt64(&cm.totalConnections, 1)
    
    // 处理连接逻辑
    buffer := make([]byte, 1024)
    for {
        n, err := conn.Read(buffer)
        if err != nil {
            break
        }
        conn.Write(buffer[:n])
    }
}

func (cm *ConnectionMonitor) startMetricsServer() {
    go func() {
        ticker := time.NewTicker(10 * time.Second)
        for range ticker.C {
            active := atomic.LoadInt64(&cm.activeConnections)
            total := atomic.LoadInt64(&cm.totalConnections)
            uptime := time.Since(cm.startTime)
            
            var m runtime.MemStats
            runtime.ReadMemStats(&m)
            
            fmt.Printf("=== TCP 连接监控 ===\n")
            fmt.Printf("活跃连接: %d\n", active)
            fmt.Printf("总连接数: %d\n", total)
            fmt.Printf("运行时间: %v\n", uptime)
            fmt.Printf("Goroutines: %d\n", runtime.NumGoroutine())
            fmt.Printf("内存使用: %.2f MB\n", float64(m.Alloc)/1024/1024)
            fmt.Println()
        }
    }()
}

重要性能指标：

指标类别	具体指标	含义	正常范围
连接状态	ESTABLISHED	正常连接数	根据业务需求
	TIME_WAIT	主动关闭的连接	< 总连接数的 20%
	CLOSE_WAIT	应用未关闭的连接	应该很少
吞吐量	Bytes/s	数据传输速率	接近带宽上限
	Packets/s	包传输速率	-
延迟	RTT	往返时间	< 100ms (内网)
	Connect Time	连接建立时间	< 1s
错误率	Retransmission	重传率	< 1%
	Connection Refused	连接拒绝率	< 0.1%

12. 在 Go 语言中如何优化 TCP 连接的性能？

连接参数优化：

func optimizedTCPServer() {
    // 创建监听器
    listener, err := net.Listen("tcp", ":8080")
    if err != nil {
        log.Fatal(err)
    }
    defer listener.Close()
    
    // 类型断言获取 TCPListener
    if tcpListener, ok := listener.(*net.TCPListener); ok {
        // 可以设置一些监听器级别的参数
        _ = tcpListener
    }
    
    for {
        conn, err := listener.Accept()
        if err != nil {
            log.Printf("Accept error: %v", err)
            continue
        }
        
        // 连接级别的优化
        if tcpConn, ok := conn.(*net.TCPConn); ok {
            optimizeTCPConnection(tcpConn)
        }
        
        go handleConnection(conn)
    }
}

func optimizeTCPConnection(conn *net.TCPConn) {
    // 1. 禁用 Nagle 算法，减少小包延迟
    conn.SetNoDelay(true)
    
    // 2. 设置合适的缓冲区大小
    conn.SetReadBuffer(64 * 1024)   // 64KB 读缓冲区
    conn.SetWriteBuffer(64 * 1024)  // 64KB 写缓冲区
    
    // 3. 设置 Keep-Alive 参数
    conn.SetKeepAlive(true)
    conn.SetKeepAlivePeriod(30 * time.Second)
    
    // 4. 设置读写超时
    conn.SetReadDeadline(time.Now().Add(30 * time.Second))
    conn.SetWriteDeadline(time.Now().Add(30 * time.Second))
}

高性能服务器模式：

// 1. 使用工作池模式
type WorkerPool struct {
    workerCount int
    jobQueue    chan net.Conn
    workers     []*Worker
}

type Worker struct {
    id       int
    jobQueue chan net.Conn
    quit     chan bool
}

func (w *Worker) start() {
    go func() {
        for {
            select {
            case conn := <-w.jobQueue:
                w.handleConnection(conn)
            case <-w.quit:
                return
            }
        }
    }()
}

func (w *Worker) handleConnection(conn net.Conn) {
    defer conn.Close()
    
    // 高效的数据处理
    reader := bufio.NewReader(conn)
    writer := bufio.NewWriter(conn)
    
    for {
        line, err := reader.ReadString('\n')
        if err != nil {
            break
        }
        
        // 处理数据
        response := processData(line)
        
        writer.WriteString(response)
        writer.Flush()
    }
}

// 2. 零拷贝优化
func zeroCopyTransfer(dst net.Conn, src *os.File) error {
    // 在 Linux 下可以使用 sendfile 系统调用
    if tcpConn, ok := dst.(*net.TCPConn); ok {
        file, err := tcpConn.File()
        if err != nil {
            return err
        }
        defer file.Close()
        
        // 使用 splice/sendfile 进行零拷贝传输
        return syscall.Sendfile(int(file.Fd()), int(src.Fd()), nil, 1024*1024)
    }
    
    // 回退到普通拷贝
    _, err := io.Copy(dst, src)
    return err
}

内存优化：

// 3. 对象池复用
var bufferPool = sync.Pool{
    New: func() interface{} {
        return make([]byte, 32*1024) // 32KB 缓冲区
    },
}

func efficientHandler(conn net.Conn) {
    defer conn.Close()
    
    // 从池中获取缓冲区
    buffer := bufferPool.Get().([]byte)
    defer bufferPool.Put(buffer) // 使用完毕后归还
    
    for {
        n, err := conn.Read(buffer)
        if err != nil {
            break
        }
        
        // 处理数据
        processBuffer(buffer[:n])
        
        // 写回响应
        conn.Write(buffer[:n])
    }
}

// 4. 批量处理优化
func batchProcessor(connections []net.Conn) {
    // 使用 epoll 或类似机制批量处理多个连接
    fds := make([]int, len(connections))
    for i, conn := range connections {
        if tcpConn, ok := conn.(*net.TCPConn); ok {
            file, _ := tcpConn.File()
            fds[i] = int(file.Fd())
        }
    }
    
    // 批量检查可读状态（简化示例）
    readyFds := epollWait(fds, 1000) // 等待最多1秒
    
    for _, fd := range readyFds {
        // 处理就绪的连接
        handleReadyConnection(fd)
    }
}

13. 如何实现一个支持百万并发连接的 TCP 服务器？

系统级优化：

# 1. 系统参数调优
# 增加文件描述符限制
echo "* soft nofile 1048576" >> /etc/security/limits.conf
echo "* hard nofile 1048576" >> /etc/security/limits.conf

# TCP 参数优化
echo 1 > /proc/sys/net/ipv4/tcp_tw_reuse
echo 1 > /proc/sys/net/ipv4/tcp_tw_recycle
echo 32768 > /proc/sys/net/core/somaxconn
echo 262144 > /proc/sys/net/core/netdev_max_backlog

# 内存参数调优
echo "1024 87380 16777216" > /proc/sys/net/ipv4/tcp_rmem
echo "1024 65536 16777216" > /proc/sys/net/ipv4/tcp_wmem

Go 程序架构：

package main

import (
    "context"
    "fmt"
    "log"
    "net"
    "runtime"
    "sync/atomic"
    "time"
)

// 百万连接服务器
type MillionConnServer struct {
    listener        net.Listener
    connCount      int64
    messageCount   int64
    workerPool     *WorkerPool
    ctx            context.Context
    cancel         context.CancelFunc
}

func NewMillionConnServer(addr string) *MillionConnServer {
    ctx, cancel := context.WithCancel(context.Background())
    
    return &MillionConnServer{
        workerPool: NewWorkerPool(runtime.NumCPU() * 2),
        ctx:        ctx,
        cancel:     cancel,
    }
}

func (s *MillionConnServer) Start(addr string) error {
    listener, err := net.Listen("tcp", addr)
    if err != nil {
        return err
    }
    s.listener = listener
    
    // 启动工作池
    s.workerPool.Start()
    
    // 启动监控
    go s.startMonitoring()
    
    // 接受连接循环
    for {
        select {
        case <-s.ctx.Done():
            return nil
        default:
            conn, err := listener.Accept()
            if err != nil {
                continue
            }
            
            // 优化连接参数
            if tcpConn, ok := conn.(*net.TCPConn); ok {
                tcpConn.SetNoDelay(true)
                tcpConn.SetKeepAlive(true)
                tcpConn.SetKeepAlivePeriod(60 * time.Second)
            }
            
            atomic.AddInt64(&s.connCount, 1)
            
            // 提交到工作池
            s.workerPool.Submit(conn)
        }
    }
}

func (s *MillionConnServer) startMonitoring() {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            connCount := atomic.LoadInt64(&s.connCount)
            msgCount := atomic.LoadInt64(&s.messageCount)
            
            var m runtime.MemStats
            runtime.ReadMemStats(&m)
            
            fmt.Printf("连接数: %d, 消息数: %d, 内存: %.2f MB, Goroutines: %d\n",
                connCount, msgCount,
                float64(m.Alloc)/1024/1024,
                runtime.NumGoroutine())
                
        case <-s.ctx.Done():
            return
        }
    }
}

// 高效的工作池实现
type WorkerPool struct {
    workerCount int
    jobQueue    chan net.Conn
    workers     []*Worker
}

func NewWorkerPool(workerCount int) *WorkerPool {
    return &WorkerPool{
        workerCount: workerCount,
        jobQueue:    make(chan net.Conn, workerCount*100), // 缓冲队列
    }
}

func (wp *WorkerPool) Start() {
    wp.workers = make([]*Worker, wp.workerCount)
    
    for i := 0; i < wp.workerCount; i++ {
        worker := &Worker{
            id:       i,
            jobQueue: wp.jobQueue,
            quit:     make(chan bool),
        }
        wp.workers[i] = worker
        worker.start()
    }
}

func (wp *WorkerPool) Submit(conn net.Conn) {
    select {
    case wp.jobQueue <- conn:
    default:
        // 队列满了，直接关闭连接
        conn.Close()
    }
}

type Worker struct {
    id       int
    jobQueue chan net.Conn
    quit     chan bool
}

func (w *Worker) start() {
    go func() {
        for {
            select {
            case conn := <-w.jobQueue:
                w.handleConnection(conn)
            case <-w.quit:
                return
            }
        }
    }()
}

func (w *Worker) handleConnection(conn net.Conn) {
    defer conn.Close()
    
    // 设置读超时
    conn.SetReadDeadline(time.Now().Add(30 * time.Second))
    
    buffer := make([]byte, 1024)
    for {
        n, err := conn.Read(buffer)
        if err != nil {
            break
        }
        
        // 简单回显
        conn.Write(buffer[:n])
        
        // 重置超时
        conn.SetReadDeadline(time.Now().Add(30 * time.Second))
    }
}

// 压力测试客户端
func stressTestClient(serverAddr string, clientCount int) {
    sem := make(chan struct{}, 1000) // 限制并发连接数
    
    for i := 0; i < clientCount; i++ {
        go func(id int) {
            sem <- struct{}{} // 获取许可
            defer func() { <-sem }() // 释放许可
            
            conn, err := net.Dial("tcp", serverAddr)
            if err != nil {
                log.Printf("客户端 %d 连接失败: %v", id, err)
                return
            }
            defer conn.Close()
            
            // 发送测试数据
            testData := fmt.Sprintf("Hello from client %d\n", id)
            conn.Write([]byte(testData))
            
            // 读取响应
            buffer := make([]byte, 1024)
            n, err := conn.Read(buffer)
            if err != nil {
                log.Printf("客户端 %d 读取失败: %v", id, err)
                return
            }
            
            log.Printf("客户端 %d 收到响应: %s", id, string(buffer[:n]))
            
            // 保持连接一段时间
            time.Sleep(time.Minute)
        }(i)
        
        // 控制连接建立速度
        if i%1000 == 0 {
            time.Sleep(100 * time.Millisecond)
        }
    }
}

func main() {
    server := NewMillionConnServer(":8080")
    
    go func() {
        log.Println("启动压力测试客户端...")
        time.Sleep(2 * time.Second)
        stressTestClient("localhost:8080", 100000) // 10万连接测试
    }()
    
    log.Println("启动百万连接服务器...")
    if err := server.Start(":8080"); err != nil {
        log.Fatal(err)
    }
}

关键优化点：

1. 内存优化：

   • 使用对象池复用缓冲区
   • 控制 Goroutine 数量
   • 及时释放不需要的资源

2. 网络优化：

   • 合理设置 TCP 参数
   • 使用批量处理
   • 实现背压控制

3. 并发优化：

   • 工作池模式限制 Goroutine 数量
   • 使用缓冲 channel 减少阻塞
   • 合理的超时机制

4. 系统优化：

   • 调整内核参数
   • 增加文件描述符限制
   • 使用合适的调度策略